package com.yunhe.transfer.mqtt;

import cn.beepower.domain.bus.MqttDataBus;
import cn.beepower.domain.bus.ThreadMode;
import cn.beepower.domain.message.BeeMessage;
import cn.beepower.domain.message.Payload;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.transfer.client.AuthorityClient;
import com.yunhe.transfer.client.ControlClient;
import com.yunhe.common.constant.CacheConsts;
import com.yunhe.common.constant.ControlConsts;
import com.yunhe.common.constant.MqttConsts;
import com.yunhe.common.model.ResultObject;
import com.yunhe.common.model.model.RunStrategyDistributeDTO;
import com.yunhe.common.util.JsonUtil;
import com.yunhe.common.util.RedisClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Component
@SuppressWarnings("ALL")
public class RunStrategyMqttHandler {

    @Autowired
    @Qualifier("MqttDataBus")
    private MqttDataBus mqttDataBus;
    @Autowired
    private RedisClient redisClient;
    @Autowired
    private ControlClient controlClient;
    @Autowired
    private AuthorityClient authorityClient;

    private static final Logger logger = LoggerFactory.getLogger(RunStrategyMqttHandler.class);

    private static final ObjectMapper mapper = JsonUtil.getObjectMapper();

    private static final Map<String, String> tokenMap = new ConcurrentHashMap<>();

    private static final int delay = 5000;

    /**
     * 发送Mqtt消息
     */
    public ResultObject send(RunStrategyDistributeDTO runStrategyDistributeDTO) throws Exception {
        long stationCode = runStrategyDistributeDTO.getStationCode();
        String topic = MqttConsts.TOPIC.RunStrategyClient.value(stationCode);
        String token = UUID.randomUUID().toString();
        synchronized (token) {
            tokenMap.put(token, token);
            runStrategyDistributeDTO.setToken(token);
            mqttDataBus.publish(topic, mapper.writeValueAsString(runStrategyDistributeDTO), 1);
            token.wait(delay);
            tokenMap.remove(token);
        }
        String key = CacheConsts.KEY.RUN_STRATEGY.value(token);
        RunStrategyDistributeDTO dataMap = redisClient.getObject(key, RunStrategyDistributeDTO.class);
        if (dataMap != null) {
            if (dataMap.getErrorCode() == 0) {
                logger.info("下发策略成功");
                return new ResultObject<>(dataMap.getResults());
            } else {
                logger.error("下发策略失败：{}", dataMap);
                return new ResultObject<>(dataMap.getErrorCode(), dataMap.getErrorMsg());
            }
        } else {
            logger.warn("下发策略失败");
            return new ResultObject<>(99, "下发策略失败");
        }
    }

    /**
     * 订阅Mqtt消息
     */
    public void subscribe(Long stationCode) {
        String topic = MqttConsts.TOPIC.RunStrategyServer.value(stationCode);
        mqttDataBus.subscribe(topic, 1, ThreadMode.Async, msg -> {
            BeeMessage message = msg.getM();
            Long code = MqttConsts.TOPIC.RunStrategyServer.getStationCode(message.getTopic());
            try {
                Payload payload = message.getPayload();
                String result = payload.getValue().toStringUtf8();
                RunStrategyDistributeDTO runStrategyDistributeDTO = mapper.readValue(result, RunStrategyDistributeDTO.class);
                if (runStrategyDistributeDTO != null) {
                    runStrategyDistributeDTO.setAction("RESPONSE");
                    if (runStrategyDistributeDTO.getType() != null && ControlConsts.RunStrategyType.Current.value().equals(runStrategyDistributeDTO.getType())) {
                        Boolean lock = redisClient.tryGetDistributedLock(code + runStrategyDistributeDTO.getName(), UUID.randomUUID().toString(), 10);
                        if (lock) {
                            //当前正在执行的策略
                            try {
                                controlClient.saveRunStrategyHistory(code, runStrategyDistributeDTO.getName(), runStrategyDistributeDTO.getControlMode(),
                                        (Map<String, Object>) runStrategyDistributeDTO.getResults());
                                runStrategyDistributeDTO.setErrorCode(0);
                            } catch (Exception e) {
                                runStrategyDistributeDTO.setErrorCode(1);
                                runStrategyDistributeDTO.setErrorMsg(e.getMessage());
                            }
                        }
                    } else {
                        //策略下发返回结果
                        String tokenObj = runStrategyDistributeDTO.getToken();
                        if (tokenObj != null) {
                            String token = tokenMap.get(tokenObj);
                            if (token != null) {
                                synchronized (token) {
                                    String key = CacheConsts.KEY.RUN_STRATEGY.value(token);
                                    redisClient.setObject(key, runStrategyDistributeDTO, CacheConsts.getDefaultTime());
                                    token.notifyAll();
                                }
                            }
                        }
                        runStrategyDistributeDTO.setErrorCode(0);
                    }
                    //向站端发送确认消息
                    mqttDataBus.publish(MqttConsts.TOPIC.RunStrategyClient.value(stationCode), mapper.writeValueAsString(runStrategyDistributeDTO), 1);
                }
            } catch (Exception e) {
                logger.error("Mqtt接收消息出错，主题：{}", message.getTopic(), e);
            }
            return null;
        });
    }
}
